package com.example.homework1

import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lbq
 * @date 2022/2/20 21:17
 * @version 1.0
 */
object MyKafkaConsumer extends Serializable {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getCanonicalName)
    val sc: SparkContext = new SparkContext(conf)

    val ssc: StreamingContext = new StreamingContext(sc, Seconds(10))
    sc.setLogLevel("warn")

    // 定义kafka相关参数
    val kafkaParams: Map[String, Object] = getKafkaConsumerParams("consumerB")
    val topics: Array[String] = Array("topic_A1")

    val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics,
        kafkaParams)
    )

    lazy val producer = new KafkaProducer[String, String](getKafkaProducerParams)
    val topic = "topic_B1"

    dStream.foreachRDD(records => {
      if (records.count > 0) {
        records.map(line => {
          val key = line.key()
          val value = line.value().split(",").map(_.replaceAll("<<<!>>>", "")).mkString(" | ")
          (key, value)
        }).foreach(line => {
          val msg: ProducerRecord[String, String] = new ProducerRecord[String, String](topic, line._1, line._2)
          producer.send(msg)
        })
      }
    })

    ssc.start()
    ssc.awaitTermination()
  }


  def getKafkaConsumerParams(groupId: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "aliyun:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )
  }

  def getKafkaProducerParams: Properties = {
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "aliyun:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    prop.put(ProducerConfig.ACKS_CONFIG, "1")
    prop
  }
}
